package com.qssoft.cqry.synchdb.config;

import com.alibaba.fastjson.support.spring.FastJsonRedisSerializer;
import com.qssoft.cqry.synchdb.adapter.MysqlBinlogAdapter;
import com.qssoft.cqry.synchdb.adapter.SynchDbAdapter;
import com.qssoft.cqry.synchdb.handler.MysqlBinlogHandler;
import com.qssoft.cqry.synchdb.handler.MysqlBinlogPositionHandler;
import com.qssoft.cqry.synchdb.handler.RedisMessageHandler;
import com.qssoft.cqry.synchdb.handler.RedisSynchHandler;
import com.qssoft.cqry.synchdb.handler.client.ClientSynchDbHandler;
import com.qssoft.cqry.synchdb.handler.server.ServerSynchDbHandler;
import com.qssoft.cqry.synchdb.pool.MysqlBinlogClient;
import com.qssoft.cqry.synchdb.pool.RedisMessagePool;
import com.qssoft.cqry.synchdb.properties.SynchDbProperties;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.boot.autoconfigure.data.redis.RedisProperties;
import org.springframework.boot.autoconfigure.jdbc.DataSourceProperties;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.boot.jdbc.DataSourceBuilder;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Import;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.data.redis.connection.RedisStandaloneConfiguration;
import org.springframework.data.redis.connection.jedis.JedisClientConfiguration;
import org.springframework.data.redis.connection.jedis.JedisConnectionFactory;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.serializer.StringRedisSerializer;
import org.springframework.jdbc.core.JdbcTemplate;
import redis.clients.jedis.JedisPoolConfig;

import javax.sql.DataSource;

/**
 * @author liju.z
 */
@Configuration
@EnableConfigurationProperties({SynchDbProperties.class, DataSourceProperties.class})
@Import({
        SynchDbConfiguration.MasterMysqlBinlogConfig.class,
        SynchDbConfiguration.SlaveMysqlBinlogConfig.class,
        SynchDbConfiguration.BaseDataSourceAndJdbcTemplate.class
})
public class SynchDbConfiguration {

    @Autowired
    private SynchDbProperties synchDbProperties;

    /**
     * 创建 RedisConnectionFactory
     *
     * @return
     */
    @Bean("SynchDbRedisConnectionFactory")
    public RedisConnectionFactory connectionFactory() {
        JedisPoolConfig poolConfig = new JedisPoolConfig();
        RedisProperties redisProperties = synchDbProperties.getRedis();
        RedisProperties.Pool pool = redisProperties.getLettuce().getPool();
        if (pool != null) {
            poolConfig.setMaxTotal(pool.getMaxActive());
            poolConfig.setMaxIdle(pool.getMaxIdle());
            poolConfig.setMaxWaitMillis(pool.getMaxWait().toMillis());
            poolConfig.setMinIdle(pool.getMinIdle());
        }
        poolConfig.setTestOnBorrow(true);
        poolConfig.setTestOnReturn(false);
        poolConfig.setTestWhileIdle(true);
        //.readTimeout(Duration.ofMillis(redisTimeout))
        JedisClientConfiguration clientConfig = JedisClientConfiguration.builder().usePooling().poolConfig(poolConfig).and().build();
        // 单点redis
        RedisStandaloneConfiguration redisConfig = new RedisStandaloneConfiguration();
        // 哨兵redis
        // RedisSentinelConfiguration redisConfig = new RedisSentinelConfiguration();
        // 集群redis
        // RedisClusterConfiguration redisConfig = new RedisClusterConfiguration();
        redisConfig.setHostName(redisProperties.getHost());
        redisConfig.setPassword(redisProperties.getPassword());
        redisConfig.setPort(redisProperties.getPort());
        redisConfig.setDatabase(redisProperties.getDatabase());
        return new JedisConnectionFactory(redisConfig, clientConfig);
    }

    /**
     * 不能用虚拟化方式，实体有些未实现Serializable接口
     * 创建 RedisTemplate<String, Object>
     *
     * @return
     */
    @Bean("sychDbRedisTemplate")
    public RedisTemplate<String, Object> redisTemplate() {
        RedisTemplate<String, Object> redisTemplate = new RedisTemplate<>();
        redisTemplate.setKeySerializer(new StringRedisSerializer());
        redisTemplate.setHashKeySerializer(new StringRedisSerializer());

        FastJsonRedisSerializer<Object> fastJsonRedisSerializer = new FastJsonRedisSerializer(Object.class);
        redisTemplate.setHashValueSerializer(fastJsonRedisSerializer);
        redisTemplate.setValueSerializer(fastJsonRedisSerializer);
        redisTemplate.setConnectionFactory(connectionFactory());
        return redisTemplate;
    }

    /**
     * 数据源
     */
    public class BaseDataSourceAndJdbcTemplate {

        @Bean("masterDataSource")
        @ConfigurationProperties(prefix = "spring.synchdb.master-db")
        public DataSource masterDataSource() {
            return DataSourceBuilder.create().build();
        }

        @Bean("masterJdbcTemplate")
        JdbcTemplate masterJdbcTemplate(@Qualifier("masterDataSource") DataSource dsOne) {
            return new JdbcTemplate(dsOne);
        }

        @Bean("slaveDataSource")
        @ConfigurationProperties(prefix = "spring.synchdb.slave-db")
        public DataSource slaveDataSource() {
            return DataSourceBuilder.create().build();
        }

        @Bean("slaveJdbcTemplate")
        JdbcTemplate slaveJdbcTemplate(@Qualifier("slaveDataSource") DataSource dsOne) {
            return new JdbcTemplate(dsOne);
        }
    }

    /**
     * master
     */
    @ConditionalOnProperty(value = "spring.synchdb.master-db.enable", matchIfMissing = true)
    public class MasterMysqlBinlogConfig {
        /**
         * redis处理
         */
        RedisSynchHandler redisSynchHandler = new RedisSynchHandler(synchDbProperties.getMasterDb(), redisTemplate());
        /**
         * binlog文件记录标记
         */
        MysqlBinlogPositionHandler mysqlBinlogPositionHandler = new MysqlBinlogPositionHandler(redisTemplate(), synchDbProperties.getApplicationName(), synchDbProperties.getMasterDb());

        /**
         * binlog链接监听器
         *
         * @return
         */
        @Bean
        public MysqlBinlogClient masterMysqlBinlogClient() {
            return new MysqlBinlogClient(synchDbProperties.getMasterDb(), mysqlBinlogPositionHandler);
        }

        /**
         * 数据库sql处理
         *
         * @return
         */
        @Bean
        @ConditionalOnMissingBean(MysqlBinlogAdapter.class)
        public MysqlBinlogAdapter masterMysqlBinlogHandler() {
            return new MysqlBinlogHandler(redisSynchHandler);
        }

        /**
         * 创建服务端同步数据处理，支持自定义
         *
         * @return
         */
        @Bean
        @ConditionalOnMissingBean(ServerSynchDbHandler.class)
        public SynchDbAdapter serverSynchDbHandler() {
            return new ServerSynchDbHandler();
        }

        /**
         * 创建服务端线程
         *
         * @param serverSynchDbHandler
         * @return
         */
        @Bean(destroyMethod = "destroy")
        public RedisMessagePool serverMessage(@Qualifier("slaveJdbcTemplate") JdbcTemplate jdbcTemplate, ServerSynchDbHandler serverSynchDbHandler) {
            RedisMessageHandler redisMessageHandler = new RedisMessageHandler(serverSynchDbHandler, jdbcTemplate, redisSynchHandler);
            return new RedisMessagePool(redisMessageHandler, serverSynchDbHandler.getClass().getName());
        }
    }

    /**
     * slave
     */
    @ConditionalOnProperty(value = "spring.synchdb.slave-db.enable", matchIfMissing = true)
    public class SlaveMysqlBinlogConfig {

        /**
         * redis处理
         */
        RedisSynchHandler redisSynchHandler = new RedisSynchHandler(synchDbProperties.getSlaveDb(), redisTemplate());
        /**
         * binlog文件记录标记
         */
        MysqlBinlogPositionHandler mysqlBinlogPositionHandler = new MysqlBinlogPositionHandler(redisTemplate(), synchDbProperties.getApplicationName(), synchDbProperties.getSlaveDb());
        /**
         * binlog监听器
         * @return
         */
        @Bean
        public MysqlBinlogClient slaveMysqlBinlogClient() {
            return new MysqlBinlogClient(synchDbProperties.getSlaveDb(), mysqlBinlogPositionHandler);
        }

        /**
         * 数据库sql处理
         *
         * @return
         */
        @Bean
        @ConditionalOnMissingBean(MysqlBinlogAdapter.class)
        public MysqlBinlogAdapter slaveMysqlBinlogHandler() {
            return new MysqlBinlogHandler(redisSynchHandler);
        }

        /**
         * 创建客服端同步数据处理，支持自定义
         *
         * @return
         */
        @Bean
        @ConditionalOnMissingBean(ClientSynchDbHandler.class)
        public SynchDbAdapter clientSynchDbHandler() {
            return new ClientSynchDbHandler();
        }

        /**
         * 创建客服端线程
         *
         * @param clientSynchDbHandler
         * @return
         */
        @Bean(destroyMethod = "destroy")
        @ConditionalOnProperty(value = "spring.synchdb.slave-db.enable", matchIfMissing = true)
        public RedisMessagePool slaveMessage(@Qualifier("slaveJdbcTemplate") JdbcTemplate jdbcTemplate, ClientSynchDbHandler clientSynchDbHandler) {
            RedisMessageHandler redisMessageHandler = new RedisMessageHandler(clientSynchDbHandler, jdbcTemplate, redisSynchHandler);
            return new RedisMessagePool(redisMessageHandler, clientSynchDbHandler.getClass().getName());
        }
    }
}